package com.lsh.oms.core.mq.kafka;

import com.alibaba.fastjson.JSON;
import com.lsh.base.lang.Exceptions;
import com.lsh.oms.core.mq.MessageProducer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.util.HashMap;
import java.util.Map;

import static org.apache.kafka.clients.producer.ProducerConfig.*;

@Component("kafka.message.producer")
public class KafkaMessageProducer implements MessageProducer, Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageProducer.class);

    private KafkaProducer<String, Object> producer;

    public KafkaMessageProducer(@Value("${kafka.bootstrap.address}") String bootstrapAddress,
                                @Value("${kafka.producer.acks:all}") String acks,
                                @Value("${kafka.producer.retries:3}") Integer retries,
                                @Value("${kafka.producer.batch.size:16384}") Integer batchSize,
                                @Value("${kafka.producer.linger.ms:1}") Integer lingerMs,
                                @Value("${kafka.producer.buffer.memory:33554432}") Integer bufferMemory) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        configs.put(ACKS_CONFIG, acks);
        configs.put(RETRIES_CONFIG, retries);
        configs.put(BATCH_SIZE_CONFIG, batchSize);
        configs.put(LINGER_MS_CONFIG, lingerMs);
        configs.put(BUFFER_MEMORY_CONFIG, bufferMemory);
        configs.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        this.producer = new KafkaProducer<>(configs);
    }

    @Override
    public void send(String topic, Object message) {
        send(topic, JSON.toJSONString(message));
    }

    @Override
    public void send(String topic, Map<String, Object> message) {
        send(topic, JSON.toJSONString(message));
    }

    @Override
    public void send(final String topic, final String message) {
        producer.send(new ProducerRecord<>(topic, message), (metadata, exception) -> {
            if (exception != null) {
                LOGGER.error("send message: {} on topic: {} failed, exception message: {}", message, topic, exception.getMessage());
                Exceptions.printStackTraceLineByLine(exception);
            } else {
                LOGGER.info("sent message: {} on topic: {} success with offset: {}", new Object[]{message, topic, metadata.offset()});
            }
        });
    }

    @Override
    public void close() {
        producer.close();
    }
}
